package com.ws.client.socketClient;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ws.common.constant.Constant;
import com.ws.common.entity.TaxiTrajectory;
import com.ws.common.enums.SocketMsgTypeEnum;
import com.ws.common.util.KafkaUtils;
import com.ws.common.util.SocketMsgUtil;
import com.ws.common.util.SocketUtil;
import com.ws.common.util.StreamUtil;
import com.ws.common.vo.RealTimeMsgEntity;
import com.ws.common.vo.SocketMsgDataVo;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.util.List;
import java.util.Optional;

/**
 * 客户端发送，服务端消息接收线程
 */
public class ClientRecvThread implements Runnable {

    private final static Logger log = LoggerFactory.getLogger(ClientRecvThread.class);

    private Socket socket;

    private volatile boolean isStop = false;

    public ClientRecvThread(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() {
        //线程终止条件： 设置标志位为 true or socket 已关闭
        InputStream inputStream = null;
        DataInputStream dataInputStream = null;
        try {
            inputStream = socket.getInputStream();
            /* dataInputStream = new DataInputStream(inputStream);*/
            int i = 0;
            StringBuffer stringBuffer = new StringBuffer();
            while (!isStop && !socket.isClosed()) {
                String msg = SocketUtil.readMsgData(inputStream);
                //根据通讯协议，解析正文json数据
                if (msg != null) {
                    String receive = msg.replaceAll("\\^","");
                    JSONObject jsonObject = JSON.parseObject(receive);
                    if (null != jsonObject) {
                        String msgStr = jsonObject.getString("data");
                        if (StringUtils.isNotEmpty(msgStr)) {
                            List<TaxiTrajectory> list = JSONArray.parseArray(msgStr, TaxiTrajectory.class);
                            for (TaxiTrajectory taxiTrajectory : list) {
                               // String JsonStr = JSONArray.toJSON(taxiTrajectory).toString();
                                String JsonStr=JSON.toJSONString(taxiTrajectory);
                                log.info("处理好的出租车轨迹json消息为:{}", JsonStr);
                                Optional.ofNullable(JsonStr).ifPresent(u -> KafkaUtils.sendMsgToKafka(u, Constant.kafka_czcgj_topic));
                            }
                        }

                    }

                  /*
                    for (String retval : msg.split("\\$")) {

                        *//*     *//**//**1,GPS和站点信息字符处理和发送到kafka*//**//*
                        //处理好的json字符串
                        String JsonStr = SocketMsgUtil.gpsAndLocation(retval);
                        log.info("处理好的json消息为:{}", JsonStr);
                       Optional.ofNullable(JsonStr).ifPresent(u -> KafkaUtils.sendMsgToKafka(JsonStr, Constant.kafka_producer_topic));
                        *//**//**2,线路信息*//**//*
                       SocketMsgUtil.routeInformation(retval);*//*
                     *//**3实时GPS站点保存到文件*//*
                     *//* SocketMsgUtil.saveFile(retval);
                            i++;*//*
                     *//**4石家庄出租车轨迹信息*//*
                        //   SocketMsgUtil.saveFile2(retval);
                    }
*/
                }
            }
        } catch (IOException e) {
            log.error("客户端接收消息发生异常");
            e.printStackTrace();
        } finally {
            this.isStop = true;
            log.info("客户端旧接收线程已摧毁");
            StreamUtil.closeInputStream(dataInputStream);
            StreamUtil.closeInputStream(inputStream);
            SocketUtil.closeSocket(socket);
            /*if (socket.isClosed()) {
                System.out.println("socket.isClosed");
            }*/
        }

    }

    public boolean getStop() {
        return isStop;
    }

    public void setStop(boolean stop) {
        isStop = stop;
    }
}
